package com.horse.cloud.framework.rabbitmq.core;


import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;



/**
 * @create: Created by intelliJ IDEA18.1.3
 * @author: bing.Pan
 * @e-mail: 15923508369@163.com
 * @gmdate: 17/05/2019　17:32　星期五 (dd/MM/YYYY HH:mm)
 * @sidesc:
 */

@Configuration
public class RabbitConfig {
    private static final Logger LOGGER = LoggerFactory.getLogger(RabbitConfig.class);

    @Autowired  private RabbitProperties rabbitProperties;

    @Bean
    @ConditionalOnMissingBean
    public RabbitProperties redisProperties(){ return new RabbitProperties(); }

    final static String EXCHANGE_NAME = "amq.topic";
    final static String QUEUE_NAME = "topic.baqgl.*.*";
    final static String ROUTING_KEY = "topic.baqgl.#";

    @Bean
    public ConnectionFactory connectionFactory() {

        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(rabbitProperties.getHost() + ":" + rabbitProperties.getPort());
        connectionFactory.setUsername(rabbitProperties.getUsername());
        connectionFactory.setPassword(rabbitProperties.getPassword());
        connectionFactory.setVirtualHost(rabbitProperties.getVirtualHost());

        connectionFactory.setPublisherConfirms(rabbitProperties.getPublisherConfirms());
        return connectionFactory;
    }


    /**
     * 因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最后一次设置
     * @return
     */
    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate template = new RabbitTemplate(connectionFactory());
        template.setMandatory(true);
        return template;
    }





    @Bean
    TopicExchange exchange() {
        return new TopicExchange(EXCHANGE_NAME);
    }

    @Bean
    public Queue queue() {
        return new Queue(QUEUE_NAME, true);
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with(ROUTING_KEY);
    }


    @Bean
    public SimpleMessageListenerContainer messageContainer() {

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        container.setQueues(queue());
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(1);
        container.setConcurrentConsumers(1);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(new ChannelAwareMessageListener() {

            public void onMessage(Message message, Channel channel) throws Exception {
                try {
                    LOGGER.info("【framework-rabbit】 consumer topic[{}]",message.getMessageProperties().getReceivedRoutingKey());
                    LOGGER.info("【framework-rabbit】 consumer messageProperties[{}],消息内容[{}]",message.getMessageProperties(),new String(message.getBody()));
                    if(message.getBody().toString().endsWith("4")){
                        int x = 1 / 0;
                    }

                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); // false只确认当前一个消息收到，true确认所有consumer获得的消息
                } catch (Exception e) {
                    e.printStackTrace();

                    if (message.getMessageProperties().getRedelivered()) {
                        LOGGER.info("【framework-rabbit】 consumer 消息已重复处理失败,拒绝再次接收");
                        channel.basicReject(message.getMessageProperties().getDeliveryTag(), true); // 拒绝消息
                    } else {
                        LOGGER.info("【framework-rabbit】 consumer 消息即将再次返回队列处理");
                        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true); // requeue为是否重新回到队列
                    }
                }
            }
        });
        return container;
    }



}
